package com.zetyun.tiger.connector.cnt;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Sets;
import com.zetyun.rt.meta.annotation.ActionMeta;
import com.zetyun.rt.sdk.action.MapAction;
import com.zetyun.rt.sdk.model.RtEvent;
import com.zetyun.rt.sdk.operator.OperatorContext;
import com.zetyun.tiger.connector.cdc.SimpleRedisContainer;
import lombok.SneakyThrows;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisSentinelConfig;

import java.util.Objects;
import java.util.Set;

import static com.zetyun.tiger.connector.op.v2.RedisOpV2Action.build;


@ActionMeta(
        id = "tiger.count-cust-cnt-with-cp",
        tags = {"redis"},
        category = "map",
        name = "CountCustCntWithCp",
        description = "统计客户经理、分支机构今日对公对私客户数",
        parameterClass = CountCustCntWithCpParam.class,
        status = ActionMeta.Status.PUBLIC
)
public class CountCustCntWithCpAction extends MapAction {


    // 公共揽存客户关系缓存
    private SimpleRedisContainer pbRelationshipDb;

    // 客户经理客户关系缓存
    private SimpleRedisContainer conntrRelationshipDb;

    // 实时数据库
    private SimpleRedisContainer rtDb;

    // 公共揽存客户关系缓存前缀
    private String pbRelationshipPrefix;

    // 客户经理客户关系缓存前缀
    private String conntrRelationshipPrefix;

    // 客户经理今日客户数
    private String cbConntrCnt;

    // 机构今日客户数
    private String cbOrgCnt;

    // 叶子机构今日开户客户号缓存
    private String orgOpacCustPrefix;

    // 叶子机构今日销户客户号缓存
    private String orgCacctCustPrefix;

    // 客户经理今日开户客户号缓存
    private String conntrOpacCustPrefix;

    // 客户经理今日销户客户号缓存
    private String conntrCacctCustPrefix;

    private CountCustCntWithCpParam param;

    private ObjectMapper om;

    private final Integer ttl = 60 * 60 * 24;  // 一天过期时间，仅 rtDb 需要设置

    @Override
    public void init(OperatorContext context) {
        super.init(context);
        om = new ObjectMapper();
        param = context.getParameters();
        pbRelationshipDb = buildContainer(param.getPbRelationshipDb());
        conntrRelationshipDb = buildContainer(param.getConntrRelationshipDb());
        rtDb = buildContainer(param.getRtDb());
        pbRelationshipPrefix = param.getPbRelationshipPrefix();

        conntrRelationshipPrefix = param.getConntrRelationshipPrefix();

//        orgCntPrefix = param.getOrgCntPrefix();
        orgOpacCustPrefix = param.getOrgOpacCustPrefix();
        orgCacctCustPrefix = param.getOrgCacctCustPrefix();
        conntrOpacCustPrefix = param.getConntrOpacCustPrefix();
        conntrCacctCustPrefix = param.getConntrCacctCustPrefix();
    }

    @Override
    public RtEvent apply(RtEvent in) throws Exception {


        String dayId = in.getField(param.getDayIdField(), String.class);
        String reaordOrgNo = in.getField(param.getRecordOrgNoField(), String.class);

        String conntrNo = in.getField(param.getConntrNoField(), String.class);
        String custNo = in.getField(param.getCustNoField(), String.class);
        String acctNo = in.getField(param.getAcctNoField(), String.class);
        String pdFlag = in.getField(param.getPdFlagField(), String.class);
        // 对公对私标识
        String pcFlag = in.getField(param.getPcFlagField(), String.class);
        String opacType = in.getField(param.getOpacFlagField(), String.class);


        dayId = ifEmpty(dayId);
        reaordOrgNo = ifEmpty(reaordOrgNo);
        conntrNo = ifEmpty(conntrNo);
        custNo = ifEmpty(custNo);
        acctNo = ifEmpty(acctNo);
        pdFlag = ifEmpty(pdFlag);
        pcFlag = ifEmpty(pcFlag);
        opacType = ifEmpty(opacType);


        // 客户经理客户关系缓存
        String conntrRelationshipCacheName = conntrRelationshipPrefix.concat(conntrNo);

        // 公共揽存缓存客户号关系
        String pbRelationshipCacheName = pbRelationshipPrefix.concat(reaordOrgNo);

        // 存储机构今日开销户客户号缓存
        String baseName = String.join(":", dayId, pcFlag, reaordOrgNo);

        String conntrBaseName = String.join(":", dayId, pcFlag, conntrNo);
        // dayId.concat(":").concat(reaordOrgNo);


        // 机构今日开户客户号缓存
        String orgOpacCustCacheName = orgOpacCustPrefix.concat(baseName);

        // 机构今日销户客户号缓存
        String orgCacctCustCacheName = orgCacctCustPrefix.concat(baseName);

        // 客户经理今日开户客户号缓存
        String conntrOpacCustCacheName = conntrOpacCustPrefix.concat(conntrBaseName);

        // 客户经理今日销户客户号缓存
        String conntrCacctCustCacheName = conntrCacctCustPrefix.concat(conntrBaseName);

        // 账号列表： 默认无账号
        Set<String> allAcctNos = Sets.newHashSet();

        switch (opacType) {
            case "01":
                // 开户
                // todo 机构今日开户客户号缓存加入客户号，其实只要是开户的话，缓存就需要加
//                rtDb.sadd(orgOpacCustCacheName, custNo, ttl);


                if ("N".equals(pdFlag)) {
                    // 客户经理也加一下
//                    rtDb.sadd(conntrOpacCustCacheName, custNo, ttl);
                    /*
                     存在客户经理，需要判断客户经理缓存中是否存在此客户号
                     */
                    String acctsStr = conntrRelationshipDb.hget(conntrRelationshipCacheName, custNo);
                    if (acctsStr != null) {
                        // 存在老账号
                        allAcctNos = strToSet(acctsStr);
                    }
                    if (allAcctNos.isEmpty()) {
                        // 没有老账户存在，则说明该客户为首次开户，此时将客户号放入对应的对公对私开户缓存客户号中
                        rtDb.sadd(orgOpacCustCacheName, custNo, ttl);
                        rtDb.sadd(conntrOpacCustCacheName, custNo, ttl);
                    }
                    allAcctNos.add(acctNo);
                    conntrRelationshipDb.hset(conntrRelationshipCacheName, custNo, setToStr(allAcctNos), null);
                } else {
                    // 公共揽存机构 客户关系新增
                    String acctsStr = pbRelationshipDb.hget(pbRelationshipCacheName, custNo);
                    if (acctsStr != null) {
                        // 存在老账号
                        allAcctNos = strToSet(acctsStr);
                    }
                    if (allAcctNos.isEmpty()) {
                        // 没有老账户存在，则说明该客户为首次开户，此时将客户号放入机构对应的对公对私开户缓存客户号中
                        rtDb.sadd(orgOpacCustCacheName, custNo, ttl);
                    }
                    allAcctNos.add(acctNo);
                    pbRelationshipDb.hset(pbRelationshipCacheName, custNo, setToStr(allAcctNos), null);
                }

                break;
            case "02":
                // 销户
                // todo 机构今日销户客户号缓存加入客户号，其实只要是销户的话，缓存就需要加
//                rtDb.sadd(orgCacctCustCacheName, custNo, ttl);

                if ("N".equals(pdFlag)) {
                    // 客户经理也加一下
//                    rtDb.sadd(conntrCacctCustCacheName, custNo, ttl);
                     /*
                     存在客户经理，需要判断客户经理缓存中是否存在此客户号
                     */
                    String acctsStr = conntrRelationshipDb.hget(conntrRelationshipCacheName, custNo);
                    if (acctsStr != null) {
                        // 存在老账号
                        allAcctNos = strToSet(acctsStr);
                    }
                    allAcctNos.remove(acctNo);
                    if (allAcctNos.size() == 0) {
                        // 删除此客户缓存
                        conntrRelationshipDb.hdel(conntrRelationshipCacheName, custNo);
                        // 此时该客户没有对应的账号存在，则为客户流失
                        rtDb.sadd(orgCacctCustCacheName, custNo, ttl);
                        rtDb.sadd(conntrCacctCustCacheName, custNo, ttl);
                    } else {
                        conntrRelationshipDb.hset(conntrRelationshipCacheName, custNo, setToStr(allAcctNos), null);
                    }
                } else {
                    // 公共揽存机构 客户关系新增
                    String acctsStr = pbRelationshipDb.hget(pbRelationshipCacheName, custNo);
                    if (acctsStr != null) {
                        // 存在老账号
                        allAcctNos = strToSet(acctsStr);
                    }
                    allAcctNos.remove(acctNo);
                }
                if (allAcctNos.size() == 0) {
                    // 删除此客户缓存
                    pbRelationshipDb.hdel(pbRelationshipCacheName, custNo);

                    // 此时该客户没有对应的账号存在，则为客户流失
                    rtDb.sadd(orgCacctCustCacheName, custNo, ttl);
                } else {
                    pbRelationshipDb.hset(pbRelationshipCacheName, custNo, setToStr(allAcctNos), null);
                }

                break;
            default:
                throw new IllegalArgumentException("不存在此类 开销户类型 {}" + opacType);
        }


        String pOrgBaseName = String.join(":", dayId, "P", reaordOrgNo);

        String cOrgBaseName = String.join(":", dayId, "C", reaordOrgNo);

        String pConntrBaseName = String.join(":", dayId, "P", conntrNo);

        String cConntrBaseName = String.join(":", dayId, "C", conntrNo);

        // P: 对私， C：对公

        // 机构今日对私开户客户号缓存
        String pOrgOpacCustCacheName = orgOpacCustPrefix.concat(pOrgBaseName);

        // 机构今日对私销户客户号缓存
        String pOrgCacctCustCacheName = orgCacctCustPrefix.concat(pOrgBaseName);

        // 机构今日对公开户客户号缓存
        String cOrgOpacCustCacheName = orgOpacCustPrefix.concat(cOrgBaseName);

        // 机构今日对公销户客户号缓存
        String cOrgCacctCustCacheName = orgCacctCustPrefix.concat(cOrgBaseName);

        // 客户经理今日对私开户客户号缓存
        String pConntrOpacCustCacheName = conntrOpacCustPrefix.concat(pConntrBaseName);

        // 客户经理今日对私销户客户号缓存
        String pConntrCacctCustCacheName = conntrCacctCustPrefix.concat(pConntrBaseName);

        // 客户经理今日对公开户客户号缓存
        String cConntrOpacCustCacheName = conntrOpacCustPrefix.concat(cConntrBaseName);

        // 客户经理今日对公销户客户号缓存
        String cConntrCacctCustCacheName = conntrCacctCustPrefix.concat(cConntrBaseName);


        // ============================================================
        // 今日机构对私客户变化数
        Long pOrgCnt = rtDb.scard(pOrgOpacCustCacheName) - rtDb.scard(pOrgCacctCustCacheName);

        // 今日机构对公客户变化数
        Long cOrgCnt = rtDb.scard(cOrgOpacCustCacheName) - rtDb.scard(cOrgCacctCustCacheName);

        // 今日客户经理对私客户变化数
        Long pConntrCnt = rtDb.scard(pConntrOpacCustCacheName) - rtDb.scard(pConntrCacctCustCacheName);

        // 今日客户经理对公客户变化数
        Long cConntrCnt = rtDb.scard(cConntrOpacCustCacheName) - rtDb.scard(cConntrCacctCustCacheName);

        in.setValue(param.getPOrgCntField(), pOrgCnt);
        in.setValue(param.getCOrgCntField(), cOrgCnt);
        in.setValue(param.getPConntrCntField(), pConntrCnt);
        in.setValue(param.getCConntrCntField(), cConntrCnt);

        return in;
    }


    @SneakyThrows
    private Set<String> strToSet(String jsonStr) {
        if (jsonStr == null || jsonStr.isEmpty()) {
            return Sets.newHashSet();
        }
        return om.readValue(jsonStr, new TypeReference<Set<String>>() {
        });
    }

    @SneakyThrows
    private String setToStr(Set<String> set) {
        return om.writeValueAsString(set);
    }

    private SimpleRedisContainer buildContainer(Integer db) {
        FlinkJedisSentinelConfig pbRelationshipDbConf =
                new FlinkJedisSentinelConfig.Builder()
                        .setMasterName(param.getMasterName())
                        .setSentinels(param.getSentinels())
                        .setPassword(param.getPassword())
                        .setDatabase(db)
                        .build();

        return build(pbRelationshipDbConf);
    }

    private String ifEmpty(String val) {
        if (Objects.isNull(val)) {
            return "";
        } else {
            return val;
        }
    }
}
